package com.dk.logc.kafka;

import com.alibaba.fastjson.JSON;
import com.dk.logc.domain.FlogEntry;
import com.dk.logc.kafka.base.Event;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;

import static com.dk.logc.config.GlobalConfig.typeCategoryMap;

/**
 * @Author 林道敬
 * @Date 2024-01-05
 * @Description
 **/
//@Service
public class KafkaBusTest {

    //@Autowired
    private KafkaTemplate<Object,Object> kafkaTemplate;

   // @PostConstruct
    public void init() throws ExecutionException, InterruptedException {
        FlogEntry flogEntry = new FlogEntry();
        flogEntry.setServerNo(1);
        Map<String,String> map = new HashMap<>();
        map.put("category",flogEntry.getCategory());
        String str = JSON.toJSONString(flogEntry);
        Event event = new Event();
        event.setHeader(map);
        event.setBody(str.getBytes());
        ProducerRecord record = new ProducerRecord("springCloudBus", event);
        kafkaTemplate.send(record);
    }

    //@KafkaListener(topics = {"springCloudBus"},groupId = "game-kafka-cluster")
    public void receiver(ConsumerRecord record) {
        Event event = (Event) record.value();
        byte[] body = event.getBody();
        Map eventMap  = event.getHeader();
        System.out.println(eventMap.get("category"));
        Class clas1s = typeCategoryMap.get(eventMap.get("category"));
        System.out.println(clas1s);
        String str = new String(body);

        System.out.println(JSON.parseObject(str,clas1s));
    }
}
